package belf.migrate.engine.job.jdbc;

import org.redisson.Redisson;
import org.redisson.api.RQueue;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

import belf.migrate.api.queue.RedisClient;

public class RedisPipe implements Pipe {
    public static final String QUEUE_NAME = "belf_jdbc";
    private static final RedissonClient redissonClient = RedisClient.getInstance().getConnection();
    private final String fullTableName;

    public RedisPipe(String fullTableName) {
        this.fullTableName = fullTableName;
    }

    @Override
    public String name() {
        return this.fullTableName;
    }

    @Override
    public void put(Object o) {
        RQueue<Object> queue = redissonClient.getQueue(QUEUE_NAME + "_" + fullTableName);
        queue.add(o);
    }

    @Override
    public Object pop() {
        RQueue<Object> queue = redissonClient.getQueue(QUEUE_NAME + "_" + fullTableName);
        return queue.poll();
    }

    @Override
    public void setTotal(long total) {
        redissonClient.getAtomicLong(QUEUE_NAME + "_" + fullTableName + "_total").set(total);
    }

    @Override
    public long getTotal() {
        return redissonClient.getAtomicLong(QUEUE_NAME + "_" + fullTableName + "_total").get();
    }

    @Override
    public void readCount(long count) {
        redissonClient.getAtomicLong(QUEUE_NAME + "_" + fullTableName + "_read").addAndGet(count);
    }

    @Override
    public long getRead() {
        return redissonClient.getAtomicLong(QUEUE_NAME + "_" + fullTableName + "_read").get();
    }

    @Override
    public long getCached() {
        RQueue<Object> queue = redissonClient.getQueue(QUEUE_NAME + "_" + fullTableName);
        return queue.size();
    }

    @Override
    public void wroteCount(long count) {
        redissonClient.getAtomicLong(QUEUE_NAME + "_" + fullTableName + "_wrote").addAndGet(count);
    }

    @Override
    public long getWrote() {
        return redissonClient.getAtomicLong(QUEUE_NAME + "_" + fullTableName + "_wrote").get();
    }
}
